package io.xxx.mario.task;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import io.xxx.mario.common.PageRequest;
import io.xxx.mario.contact.ContactMapper;
import io.xxx.mario.room.RoomMapper;
import io.xxx.mario.task.api.CreateTaskRequest;
import io.xxx.mario.task.api.UpdateTaskRequest;
import io.xxx.mario.task.job.OncePushJob;
import io.xxx.mario.task.receive.Receiver;
import io.xxx.mario.task.receive.ReceiverSelector;
import io.xxx.mario.task.receive.ReceiverSelectorFactory;
import io.xxx.mario.task.talk.TalkerFactory;
import io.xxx.mario.task.talk.TalkerManager;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.convert.ConversionService;
import org.springframework.data.util.Pair;
import org.springframework.lang.NonNull;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import javax.annotation.CheckReturnValue;
import java.sql.Date;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.time.LocalDateTime;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;

@Slf4j
@Service
public class TaskService {

    private static final NumberFormat NUMBER_FORMAT = new DecimalFormat("#,##0.00");

    private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    private static final ZoneOffset ZONE_OFFSET = OffsetDateTime.now().getOffset();

    @Value("${task.message.room.buffer.size:10000}")
    private int bufferSize;

    @Resource
    private TalkerFactory talkerFactory;

    @Resource
    private Map<String, Receiver.Filter> receiverFilters;

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private PushRecordMapper pushRecordMapper;

    @Resource
    private PushDetailMapper pushDetailMapper;

    @Resource
    private RoomMapper roomMapper;

    @Resource
    private ContactMapper contactMapper;

    @Resource
    private TaskMapper taskMapper;

    @Resource
    private TaskMessageMapper taskMessageMapper;

    @Resource
    private TaskRoomMapper taskRoomMapper;

    @Resource
    private TaskReceiverMapper taskReceiverMapper;

    @Resource
    private ConversionService conversionService;

    @Resource
    private ReceiverSelectorFactory receiverSelectorFactory;

    @Resource
    private TransactionTemplate transactionTemplate;

    @Autowired
    private Scheduler scheduler;

    @Resource
    private TalkerManager talkerManager;

    private final ThreadLocalRandom random = ThreadLocalRandom.current();

    public Page<Task> getPage(PageRequest pageRequest, String name) {
        LambdaQueryWrapper<Task> wrapper = new LambdaQueryWrapper<>();
        if (StringUtils.hasText(name)) {
            wrapper.like(Task::getName, name);
        }
        Page<Task> page = new Page<>(pageRequest.getPageNo(), pageRequest.getPageSize());
        return taskMapper.selectPage(page, wrapper);
    }

    public Task get(Long taskId) {
        return taskMapper.selectById(taskId);
    }

    @Transactional
    public Long create(CreateTaskRequest request) throws SchedulerException {
        request.validate();
        Task task = conversionService.convert(request, Task.class);
        if (task == null) {
            throw new IllegalArgumentException();
        }
        taskMapper.insert(task);
        taskMessageMapper.insertAll(task.getMessages());
        taskRoomMapper.insertAll(task.getRooms());

        Class<OncePushJob> jobClass = OncePushJob.class;
        JobDetail jobDetail = JobBuilder.newJob(jobClass)
                .withIdentity(jobClass.getSimpleName(), "TASK")
                .usingJobData("taskId", task.getId())
                .build();
        TriggerBuilder<Trigger> triggerBuilder = TriggerBuilder.newTrigger()
                .withIdentity(jobClass.getSimpleName() + "Trigger", "TASK");
        if (task.isStartNow()) {
            triggerBuilder.startNow();
        } else {
            triggerBuilder.startAt(Date.from(task.getStartTime().toInstant(ZONE_OFFSET)));
        }
        scheduler.scheduleJob(jobDetail, triggerBuilder.build());
        return task.getId();
    }

    @Transactional
    public void update(UpdateTaskRequest request) {
        request.validate();
        Task task = conversionService.convert(request, Task.class);
        if (task == null) {
            throw new IllegalArgumentException();
        }

        taskMapper.updateById(task);
        taskMessageMapper.delete(getWrapper(TaskMessage::getTaskId, task.getId()));
        taskMessageMapper.insertAll(task.getMessages());
        taskRoomMapper.delete(getWrapper(TaskRoom::getTaskId, task.getId()));
        taskRoomMapper.insertAll(task.getRooms());
    }

    private <T> LambdaUpdateWrapper<T> getWrapper(SFunction<T, Long> function, Long taskId) {
        return new LambdaUpdateWrapper<T>()
                .eq(function, taskId);
    }

    /**
     * 任务到达执行时间后启动任务。
     */
    public void initialize(Task task) {
        RLock lock = redissonClient.getLock("TASK:" + task.getId() + ":INIT");
        try {
            if (lock.tryLock()) {
                if (task.getStatus() == TaskStatus.NEW) {
                    task.setStatus(TaskStatus.INITIALIZING);
                    taskMapper.updateById(task);

                    List<TaskMessage> messages = getTaskMessages(task);
                    SelectMode selectMode = task.getSelectMode();
                    ReceiverSelector selector = receiverSelectorFactory.getSelector(selectMode);
                    int receiverCount = 0;
                    CountDownLatch latch = new CountDownLatch(1);
                    selector.select(task)
                            .filter(receiver -> {
                                Collection<Receiver.Filter> filters = receiverFilters.values();
                                for (Receiver.Filter filter : filters) {
                                    boolean result = filter.support(receiver) && filter.doFilter(receiver);
                                    if (!result)
                                        return false;
                                }
                                return true;
                            })
                            .buffer(bufferSize)
                            .map(receivers -> {
                                List<PushDetail> details = buildPushDetail(task, messages, receivers);
                                List<PushRecord> records = buildPushRecord(task, messages, receivers);
                                return Pair.of(details, records);
                            })
                            .doOnError(throwable -> log.error("任务初始化异常", throwable))
                            .doOnComplete(latch::countDown)
                            .subscribe(pair -> {
                                List<PushDetail> details = pair.getFirst();
                                List<PushRecord> records = pair.getSecond();
                                insertAll(details, records);

                            });
                    latch.await();

                    task.setReceiverCount(receiverCount);
                    task.setStatus(TaskStatus.RUNNABLE);
                    taskMapper.updateById(task);
                }
            }
        } catch (InterruptedException e) {
            log.error("任务初始化异常", e);
            throw new RuntimeException(e);
        } finally {
            if (lock.isLocked()) {
                lock.unlock();
            }
        }
    }

    private void insertAll(List<PushDetail> details, List<PushRecord> records) {
        transactionTemplate.executeWithoutResult(transactionStatus -> {
            pushDetailMapper.insertAll(details);
            pushRecordMapper.insertAll(records);
        });
    }

    private <T extends Receiver> List<PushDetail> buildPushDetail(Task task, List<TaskMessage> messages, List<T> receivers) {
        if (ObjectUtils.isEmpty(messages) || ObjectUtils.isEmpty(receivers)) {
            return Collections.emptyList();
        }
        LocalDateTime now = LocalDateTime.now();
        return receivers.stream()
                .flatMap(receiver -> messages.stream()
                        .map(message -> {
                            PushDetail it = new PushDetail();
                            it.setId(IdWorker.getId());
                            it.setTaskId(task.getId());
                            it.setReceiverType(receiver.type());
                            it.setReceiverId(receiver.id());
                            it.setMessageCount(messages.size());
                            it.setFailedCount(0);
                            it.setCreatedTime(now);
                            it.setUpdatedTime(now);
                            return it;
                        }))
                .toList();
    }

    private <T extends Receiver> List<PushRecord> buildPushRecord(Task task, List<TaskMessage> messages, List<T> receivers) {
        if (ObjectUtils.isEmpty(messages) || ObjectUtils.isEmpty(receivers)) {
            return Collections.emptyList();
        }
        LocalDateTime now = LocalDateTime.now();
        return receivers.stream()
                .flatMap(receiver -> messages.stream()
                        .map(message -> {
                            PushRecord it = new PushRecord();
                            it.setId(IdWorker.getId());
                            it.setTaskId(task.getId());
                            it.setReceiverType(receiver.type());
                            it.setReceiverId(receiver.id());
                            it.setMessageSerialNo(message.getSerialNo());
                            it.setTaskMessageId(message.getMessageId());
                            it.setMessageType(message.getMessageType());
                            it.setPushStatus(PushStatus.UN_PUSHED);
                            it.setCreatedTime(now);
                            it.setUpdatedTime(now);
                            return it;
                        }))
                .toList();
    }

    /**
     * 推送消息：
     * <p>
     * 1. 查询未推送的群
     * <p>
     * 2. 根据占比计算获取一个talker实现
     * <p>
     * 3. 发送消息
     * <p>
     * 消息推送记录走接口上报
     */
    // TODO 使用协程在应用启动时就创建对应数量的机器人，使用PG的数据监听来控制机器人的上下线
    public void push(@NonNull Task task) throws InterruptedException {
        List<TaskReceiver> taskReceivers = getUnPushedReceivers(task);
        List<TaskMessage> messages = getTaskMessages(task);
        for (TaskReceiver taskReceiver : taskReceivers) {
            Receiver receiver = getReceiver(taskReceiver);
            talkerManager.say(task, receiver, messages);
        }
    }

    @CheckReturnValue
    private List<TaskReceiver> getUnPushedReceivers(@NonNull Task task) {
        LambdaQueryWrapper<TaskReceiver> wrapper = new LambdaQueryWrapper<TaskReceiver>()
                .eq(TaskReceiver::getTaskId, task.getId())
                .eq(TaskReceiver::getStatus, PushStatus.UN_PUSHED);
        return taskReceiverMapper.selectList(wrapper);
    }

    private Receiver getReceiver(TaskReceiver taskReceiver) {
        if (taskReceiver.getReceiverType() == ReceiverType.ROOM) {
            return roomMapper.selectById(taskReceiver.getReceiverId());
        } else {
            return contactMapper.selectById(taskReceiver.getReceiverId());
        }
    }

    private List<TaskMessage> getTaskMessages(Task task) {
        LambdaQueryWrapper<TaskMessage> wrapper = new LambdaQueryWrapper<TaskMessage>()
                .eq(TaskMessage::getTaskId, task.getId());
        return taskMessageMapper.selectList(wrapper);
    }

    private void updateProcess(Task task, int i) {
        if (i % 1_000 == 0) {
            double process = 1.0 * i / task.getReceiverCount();
            task.setProcess(process);
            taskMapper.updateById(task);
            log.info("消息[{}:{}]推送进度[{}%]", task.getId(), task.getName(), NUMBER_FORMAT.format(process));
        }
    }
}
